//! Wraps a blocking queue interface as an async queue interface.

use std::sync::Arc;

use super::errors::*;
use super::reader_interface::{AsyncQueueReader, QueueReader, QueueView};

pub struct AsyncQueueWrapper<Q: QueueReader> {
    handle: tokio::runtime::Handle,
    inner: Arc<Q>,
}

impl<Q: QueueReader> AsyncQueueWrapper<Q> {
    pub fn new(handle: tokio::runtime::Handle, inner: Arc<Q>) -> Self {
        Self { handle, inner }
    }
}

impl<Q: QueueReader + Sync + Send + 'static> AsyncQueueReader for AsyncQueueWrapper<Q> {
    fn get_view(
        &self,
    ) -> impl futures::prelude::Future<Output = Result<QueueView, Error>> + 'static {
        let handle = self.handle.clone();
        let inner = self.inner.clone();
        async move {
            handle
                .spawn_blocking(move || inner.get_view())
                .await
                .map_err(|_| Error::LlmqPanic)?
        }
    }

    fn get_entry(
        &self,
        idx: u64,
    ) -> impl futures::prelude::Future<
        Output = Result<Option<aspect_core::message_queue::QueueEntry>, Error>,
    > + 'static {
        let handle = self.handle.clone();
        let inner = self.inner.clone();
        async move {
            handle
                .spawn_blocking(move || inner.get_entry(idx))
                .await
                .map_err(|_| Error::LlmqPanic)?
        }
    }

    fn consume_upto(
        &self,
        next_read_idx: u64,
    ) -> impl futures::prelude::Future<Output = Result<(), Error>> + 'static {
        let handle = self.handle.clone();
        let inner = self.inner.clone();
        async move {
            handle
                .spawn_blocking(move || inner.consume_upto(next_read_idx))
                .await
                .map_err(|_| Error::LlmqPanic)?
        }
    }
}
